package com.atguigu.gmall.realtime.apps

import java.time.LocalDateTime
import java.time.format.DateTimeFormatter

import com.alibaba.fastjson.JSON
import com.atguigu.gmall.contants.GmallConstants
import com.atguigu.gmall.realtime.bean.{OrderDetail, OrderInfo, SaleDetail, UserInfo}
import com.atguigu.gmall.realtime.utils.{MykafkaUtil, RedisUtil}
import com.google.gson.Gson
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import redis.clients.jedis.Jedis

import scala.collection.mutable.ListBuffer

/**
 * Date:2021/3/26
 * Author:csw
 * Description:
 */
object SaleApp extends BaseApp {
  override val appName: String = "Sale"
  override val seound: Int = 10

  def main(args: Array[String]): Unit = {
    runApp{
      val orderInfods: InputDStream[ConsumerRecord[String, String]] = MykafkaUtil.getKafkaStream(GmallConstants.KAFKA_TOPIC_ORDER_INFO, ssc)
      val orderDetailds: InputDStream[ConsumerRecord[String, String]] = MykafkaUtil.getKafkaStream(GmallConstants.KAFKA_TOPIC_ORDER_DETAIL, ssc)

      val orderInfoKV: DStream[(String, OrderInfo)] = orderInfods.map(record => {
        val orderInfo: OrderInfo = JSON.parseObject(record.value(), classOf[OrderInfo])
        val formatter: DateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd")
        val formatter1: DateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")
        val dateTime: LocalDateTime = LocalDateTime.parse(orderInfo.create_time,formatter1)
        orderInfo.create_date = dateTime.format(formatter)
        orderInfo.create_hour = dateTime.getHour+""
        orderInfo.consignee_tel = orderInfo.consignee_tel.replaceAll("(\\d{3})\\d{4}(\\d{4})", "$1****$2")
        (orderInfo.id, orderInfo)
      })

      val detailKV: DStream[(String, OrderDetail)] = orderDetailds.map(record => {
        val detail: OrderDetail = JSON.parseObject(record.value(), classOf[OrderDetail])
        (detail.order_id, detail)
      })

      val jds: DStream[(String, (Option[OrderInfo], Option[OrderDetail]))] = orderInfoKV.fullOuterJoin(detailKV)

      jds.mapPartitions(iter=>{
        val client: Jedis = RedisUtil.getJedisClient
        val gson = new Gson()
        val details: ListBuffer[SaleDetail]= new ListBuffer[SaleDetail]()
        iter.foreach{
          case (orderid,(orderInfoOP,orderDetailOP))=>{
            val orderInfo: OrderInfo = orderInfoOP.get
            val detail: OrderDetail = orderDetailOP.get

            if(orderInfo!=null){
              if (detail!=null){
                val saleDetail= new SaleDetail(orderInfo, detail)
                details.append(saleDetail)
              }else{
                val str: String = client.get("orderDetail" + orderInfo.id)
                if(str!=null){
                  val reDetail: OrderDetail = JSON.parseObject(str, classOf[OrderDetail])
                  val saleDetail = new SaleDetail(orderInfo, reDetail)
                  details.append(saleDetail)
                }
              }
              val str: String = gson.toJson(orderInfo)
              client.setex("orderInfo"+orderInfo.id,500,str)
            }else{
              val str: String = client.get("orderInfo" + detail.order_id)
              if(str!=null){
                val reUserInfo= JSON.parseObject(str, classOf[OrderInfo])
                val saleDetail = new SaleDetail(reUserInfo, detail)
                details.append(saleDetail)
              }else{
                val str1: String = gson.toJson(detail)
                client.setex("orderDetail"+detail.id,500,str1)
              }
            }

          }
        }



        client.close();
        null
      })

    }
  }
}



